package io.openmessaging.storage.dledger;

import com.alibaba.fastjson.JSON;
import io.openmessaging.storage.dledger.entry.DLedgerEntry;
import io.openmessaging.storage.dledger.exception.DLedgerException;
import io.openmessaging.storage.dledger.protocol.AppendEntryResponse;
import io.openmessaging.storage.dledger.protocol.DLedgerResponseCode;
import io.openmessaging.storage.dledger.protocol.PushEntryRequest;
import io.openmessaging.storage.dledger.protocol.PushEntryResponse;
import io.openmessaging.storage.dledger.statemachine.StateMachineCaller;
import io.openmessaging.storage.dledger.store.DLedgerMemoryStore;
import io.openmessaging.storage.dledger.store.DLedgerStore;
import io.openmessaging.storage.dledger.store.file.DLedgerMmapFileStore;
import io.openmessaging.storage.dledger.utils.DLedgerUtils;
import io.openmessaging.storage.dledger.utils.Pair;
import io.openmessaging.storage.dledger.utils.PreConditions;
import io.openmessaging.storage.dledger.utils.Quota;

import java.util.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

//日志复制（日志转发）
public class DLedgerEntryPusher {

    private static Logger logger = LoggerFactory.getLogger(DLedgerEntryPusher.class);

    // DLedger多副本相关配置
    public DLedgerConfig dLedgerConfig;

    // 存储实现类
    public DLedgerStore dLedgerStore;

    // 节点状态机
    public final MemberState memberState;

    // RPC服务实现类，用于与集群内的其他节点进行网络通信
    public DLedgerRpcService dLedgerRpcService;

    // <term , <peerId , indexId 从节点已存储日志序号>> 每个节点基于投票轮次的水位线标记
    // 即以投票轮次为维度，存储 复制组内每一个节点当前已存储（已追加到 PageCache)）的数据的日志序号，
    // 其主要作用是判断已提交日志的序号用于计算一条日志是否已被超过半数的节点存储
    //这个数据结构是进行日志复制仲裁的关键，即主节点收到从节点的日志复制响应后，会将复制进度存储在该数据结构中
    public Map<Long, ConcurrentMap<String, Long>> peerWaterMarksByTerm = new ConcurrentHashMap<>();

    // 等待 ACK Map 用于存放日志追加请求的响应结果（Future模式）
    //<term , indexId , future>
    public Map<Long, ConcurrentMap<Long, TimeoutFuture<AppendEntryResponse>>> pendingAppendResponsesByTerm = new ConcurrentHashMap<>();

    // 日志接收处理线程，当节点为从节点时激活
    // 从节点上开启的线程，用于接收主节点的推送请求（append、commit、append）
    public EntryHandler entryHandler;

    //日志追加ACK投票仲裁线程，当前节点为主节点时激活
    //主节点上的日志复制结果仲裁器，用于判断日志是否可提交
    public QuorumAckChecker quorumAckChecker;

    //日志请求转发器，负责向从节点转发日志，主节点为每一个 从节点构建一个 EntryDispatcher
    public Map<String, EntryDispatcher> dispatcherMap = new HashMap<>();

    //
    private Optional<StateMachineCaller> fsmCaller;

    //
    public DLedgerEntryPusher(DLedgerConfig dLedgerConfig, MemberState memberState, DLedgerStore dLedgerStore, DLedgerRpcService dLedgerRpcService) {
        this.dLedgerConfig = dLedgerConfig;
        this.memberState = memberState;
        this.dLedgerStore = dLedgerStore;
        this.dLedgerRpcService = dLedgerRpcService;

        for (String peer : memberState.getPeerMap().keySet()) {
            if (!peer.equals(memberState.getSelfId())) {
                //为每一个从节点创建一个EntryDispatcher转发线程，即每一个从节点的日志转发相互不干扰
                dispatcherMap.put(peer, new EntryDispatcher(peer, logger));
            }
        }
        //从节点启动
        this.entryHandler = new EntryHandler(logger);
        this.quorumAckChecker = new QuorumAckChecker(logger);
        this.fsmCaller = Optional.empty();
    }

    // 开启日志推送的startup()方法
    public void startup() {
        // 从节点才会被激活
        // 从节点收到 Leader 节点推送的日志并存储，然后向 Leader 节点汇报日志复制结果
        entryHandler.start();
        // Leader 节点对日志复制结果进行仲裁，如果成功存储该条日志的节点超过半数节点，则向客户端返回写入成功
        quorumAckChecker.start();
        //主节点 启动 各个 从节点之间的转发器
        for (EntryDispatcher dispatcher : dispatcherMap.values()) {
            //Leader节点将日志推送到从节点
            dispatcher.start();
        }
    }

    public void shutdown() {
        entryHandler.shutdown();
        quorumAckChecker.shutdown();
        for (EntryDispatcher dispatcher : dispatcherMap.values()) {
            dispatcher.shutdown();
        }
    }

    public void registerStateMachine(final Optional<StateMachineCaller> fsmCaller) {
        this.fsmCaller = fsmCaller;
    }

    //DLedgerEntryPusher 类的
    public CompletableFuture<PushEntryResponse> handlePush(PushEntryRequest request) throws Exception {
        return entryHandler.handlePush(request);
    }

    private void checkTermForWaterMark(long term, String env) {
        //当前表中没有存在 本轮次
        if (!peerWaterMarksByTerm.containsKey(term)) {
            //添加就完事了
            logger.info("Initialize the watermark in {} for term={}", env, term);
            ConcurrentMap<String, Long> waterMarks = new ConcurrentHashMap<>();
            //只是获得 peerId
            for (String peer : memberState.getPeerMap().keySet()) {
                //indexId  条目Id设置成 -1
                waterMarks.put(peer, -1L);
            }
            //添加就完事了
            peerWaterMarksByTerm.putIfAbsent(term, waterMarks);
        }
    }

    private void checkTermForPendingMap(long term, String env) {
        //不存在当前 轮次就添加
        if (!pendingAppendResponsesByTerm.containsKey(term)) {
            logger.info("Initialize the pending append map in {} for term={}", env, term);
            pendingAppendResponsesByTerm.putIfAbsent(term, new ConcurrentHashMap<>());
        }
    }

    /**
     * @param term   轮次
     * @param peerId 节点Id
     * @param index  条目Id
     */
    private void updatePeerWaterMark(long term, String peerId, long index, String from) {
        //logger.info("updatePeerWaterMark : from:{} , term:{} , peerId : {} , index: {} ", from, term, peerId, index);
        // System.out.printf("updatePeerWaterMark : from:{%s} , term:{%d} , peerId : {%s} , index: {%d} ", from, term, peerId, index);
        //System.out.println();
        //加锁
        synchronized (peerWaterMarksByTerm) {
            //更新 或者 添加
            checkTermForWaterMark(term, "updatePeerWaterMark");
            //更新 一个轮次下的 相对节点的 日志Id
            if (peerWaterMarksByTerm.get(term).get(peerId) < index) {
                //小于的话 就更新 说明 已经追加成功了
                peerWaterMarksByTerm.get(term).put(peerId, index);
            }
        }
    }

    public long getPeerWaterMark(long term, String peerId) {
        synchronized (peerWaterMarksByTerm) {
            checkTermForWaterMark(term, "getPeerWaterMark");
            return peerWaterMarksByTerm.get(term).get(peerId);
        }
    }

    //判断推送队列是否已满（DLedgerEntryPusher的isPendingFull()方法）
    public boolean isPendingFull(long currTerm) {
        checkTermForPendingMap(currTerm, "isPendingFull");
        //pendingAppendResponsesByTerm的数据存储格式为
        // Map< Long/*投票轮次*/, ConcurrentMap <Long/*日志序号*/, TimeoutFuture<AppendEntryResponse>>>，
        // 该方法表示每一个投票轮次积压（未提交）的日志数量默认不能超过10000条，(就是在投票选举过程中 , 收到的消息请求暂存)
        // 可通过配置参数maxPendingRequestsNum来改变默认值，即队列的长度默认为10000
        //检测当前等待从节点返回结果的个数是否超过其最大请求数量
        return pendingAppendResponsesByTerm.get(currTerm).size() > dLedgerConfig.getMaxPendingRequestsNum();
    }

    /**
     * @param entry       数据
     * @param isBatchWait 默认 true
     * @return future
     */
    public CompletableFuture<AppendEntryResponse> waitAck(DLedgerEntry entry, boolean isBatchWait) {
        //第一步：Leader 节点首先更新自身的水位线 , 确实 自己更新自己
        updatePeerWaterMark(entry.getTerm(), memberState.getSelfId(), entry.getIndex(), "waitAck");
        //假如成员就 1 个  , 说明 那就是自己
        if (memberState.getPeerMap().size() == 1) {
            AppendEntryResponse response = new AppendEntryResponse();
            response.setGroup(memberState.getGroup());
            response.setLeaderId(memberState.getSelfId());
            response.setIndex(entry.getIndex());
            response.setTerm(entry.getTerm());
            response.setPos(entry.getPos());
            if (isBatchWait) {
                //直接返回 ack
                return BatchAppendFuture.newCompletedFuture(entry.getPos(), response);
            }
            return AppendFuture.newCompletedFuture(entry.getPos(), response);
        }
        //集群中存在很多成员
        else {
            //在挂起队列中 判断 没有就添加 当前轮次
            checkTermForPendingMap(entry.getTerm(), "waitAck");
            //推送结果
            AppendFuture<AppendEntryResponse> future;
            //默认 true  说明是批量推送
            if (isBatchWait) {
                //最大 2.5 秒 等待
                future = new BatchAppendFuture<>(dLedgerConfig.getMaxWaitAckTimeMs());
            } else {
                //最大 2.5 秒 等待
                future = new AppendFuture<>(dLedgerConfig.getMaxWaitAckTimeMs());
            }
            //设置数据所在物理地址
            future.setPos(entry.getPos());
            //更新 并获得 当前轮次&& 当前日志Id的旧数据
            CompletableFuture<AppendEntryResponse> old;
            //最大容量是 10 * 1000 , 此时 放入就完事了 , 肯定还有另外一个线程去这个集合中获取元素来执行
            //todo  写入 leader 中的挂起队列中
            old = pendingAppendResponsesByTerm.get(entry.getTerm()).put(entry.getIndex(), future);
            //如果有旧数据 那么就说明 之前还在等待
            if (old != null) {
                logger.warn("[MONITOR] get old wait at index={}", entry.getIndex());
            }
            //第三步：这里是一种通用的异步编程技巧，Leader节点需要等待从节点复制完数据，通常会返回给客户端一个Future对象
            //客户端可以调用该Future的get()方法同步等待结果，而服务端会将日志序号作为键，
            // Future作为值存储在内存（ConcurrentMap）中。当结果异步返回后，
            // 需要根据日志序号找到对应的Future对象，最后填充结果，以便客户端被唤醒，从而得到响应结果
            return future;
        }
    }

    //该方法主要就是遍历转发器并唤醒
    public void wakeUpDispatchers() {
        for (EntryDispatcher dispatcher : dispatcherMap.values()) {
            dispatcher.wakeup();
        }
    }

    /**
     * Complete the TimeoutFuture in pendingAppendResponsesByTerm (CurrentTerm, index).
     * Called by statemachineCaller when a committed entry (CurrentTerm, index) was applying to statemachine done.
     *
     * @return true if complete success
     */
    public boolean completeResponseFuture(final long index) {
        final long term = this.memberState.currTerm();
        final Map<Long, TimeoutFuture<AppendEntryResponse>> responses = this.pendingAppendResponsesByTerm.get(term);
        if (responses != null) {
            CompletableFuture<AppendEntryResponse> future = responses.remove(index);
            if (future != null && !future.isDone()) {
                logger.info("Complete future, term {}, index {}", term, index);
                AppendEntryResponse response = new AppendEntryResponse();
                response.setGroup(this.memberState.getGroup());
                response.setTerm(term);
                response.setIndex(index);
                response.setLeaderId(this.memberState.getSelfId());
                response.setPos(((AppendFuture) future).getPos());
                future.complete(response);
                return true;
            }
        }
        return false;
    }

    /**
     * Check responseFutures timeout from {beginIndex} in currentTerm
     * 处理 已经超过半数的 位移之后的 挂起请求
     * 70 -> 处理 71开始的请求
     */
    public void checkResponseFuturesTimeout(final long beginIndex) {
        final long term = this.memberState.currTerm();
        final Map<Long, TimeoutFuture<AppendEntryResponse>> responses = this.pendingAppendResponsesByTerm.get(term);
        if (responses != null) {
            for (long i = beginIndex; i < Integer.MAX_VALUE; i++) {
                TimeoutFuture<AppendEntryResponse> future = responses.get(i);
                //为空 什么也不做 可能处理完了
                if (future == null) {
                    break;
                }
                //超时了 返回客户单
                else if (future.isTimeOut()) {
                    AppendEntryResponse response = new AppendEntryResponse();
                    response.setGroup(memberState.getGroup());
                    response.setCode(DLedgerResponseCode.WAIT_QUORUM_ACK_TIMEOUT.getCode());
                    response.setTerm(term);
                    response.setIndex(i);
                    response.setLeaderId(memberState.getSelfId());
                    future.complete(response);
                } else {
                    //没有超时 继续等待主节点转发
                    break;
                }
            }
        }
    }

    /**
     * Check responseFutures elapsed before {endIndex} in currentTerm
     */
    private void checkResponseFuturesElapsed(final long endIndex) {
        final long currTerm = this.memberState.currTerm();
        final Map<Long, TimeoutFuture<AppendEntryResponse>> responses = this.pendingAppendResponsesByTerm.get(currTerm);
        for (Map.Entry<Long, TimeoutFuture<AppendEntryResponse>> futureEntry : responses.entrySet()) {
            //处理  小于 endIndex 之前的 客户端请求
            if (futureEntry.getKey() < endIndex) {
                AppendEntryResponse response = new AppendEntryResponse();
                response.setGroup(memberState.getGroup());
                response.setTerm(currTerm);
                response.setIndex(futureEntry.getKey());
                //结束 么说成没成啊
                //response.setCode(0);
                response.setLeaderId(memberState.getSelfId());
                response.setPos(((AppendFuture) futureEntry.getValue()).getPos());
                futureEntry.getValue().complete(response);
                responses.remove(futureEntry.getKey());
            }
        }
    }

    //long term  : 主节点当前的投票轮次
    //long newCommittedIndex: 主节点发送日志复制请求时的已提交日志序号
    private void updateCommittedIndex(final long term, final long committedIndex, String node) {
        dLedgerStore.updateCommittedIndex(term, committedIndex, node);
        this.fsmCaller.ifPresent(caller -> caller.onCommitted(committedIndex));
    }

    /**
     * This thread will check the quorum index and complete the pending requests.
     * 主节点-日志复制仲裁
     * Raft 协议判断一条日志写入成功的标准是集群中超过半数的节点存储了该日志，
     * Leader节点首先存储数据，然后异步向它所有的从节点推送日志。
     * 不需要所有的从节点都返回日志追加成功才认为是成功写入，
     * 故Leader节点需要对返回结果进行仲裁，这部分功能主要由 QuorumAckChecker 实现
     */
    private class QuorumAckChecker extends ShutdownAbleThread {

        public QuorumAckChecker(Logger logger) {
            super("QuorumAckChecker-" + memberState.getSelfId(), logger);
        }

        //上次打印水位线的时间戳，单位为ms
        private long lastPrintWatermarkTimeMs = System.currentTimeMillis();
        //上次检测泄露的时间戳，单位为ms
        private long lastCheckLeakTimeMs = System.currentTimeMillis();
        //已投票仲裁的日志序号
        private long lastQuorumIndex = -1;

        @Override
        public void doWork() {
            try {
                //第一步：主要是输出日志相关日志。如果距上一次打印日志的时间超过3s，
                // 则输出当前的term、ledgerBegin、ledgerEnd、committed、peerWaterMarksByTerm等日志
                if (DLedgerUtils.elapsed(lastPrintWatermarkTimeMs) > 3000) {
                    if (DLedgerEntryPusher.this.fsmCaller.isPresent()) {
                        final long lastAppliedIndex = DLedgerEntryPusher.this.fsmCaller.get().getLastAppliedIndex();
                        logger.info("[{}][{}] term={} ledgerBegin={} ledgerEnd={} committed={} watermarks={} appliedIndex={}", memberState.getSelfId(), memberState.getRole(), memberState.currTerm(), dLedgerStore.getLedgerBeginIndex(), dLedgerStore.getLedgerEndIndex(), dLedgerStore.getCommittedIndex(), JSON.toJSONString(peerWaterMarksByTerm), lastAppliedIndex);
                    } else {
                        logger.info("[{}][{}] term={} ledgerBegin={} ledgerEnd={} committed={} watermarks={}", memberState.getSelfId(), memberState.getRole(), memberState.currTerm(), dLedgerStore.getLedgerBeginIndex(), dLedgerStore.getLedgerEndIndex(), dLedgerStore.getCommittedIndex(), JSON.toJSONString(peerWaterMarksByTerm));
                    }
                    lastPrintWatermarkTimeMs = System.currentTimeMillis();
                }

                //第二步：如果当前节点不是Leader，直接返回
                if (!memberState.isLeader()) {
                    waitForRunning(1);
                    return;
                }

                //当前轮次
                long currTerm = memberState.currTerm();
                checkTermForPendingMap(currTerm, "QuorumAckChecker");
                checkTermForWaterMark(currTerm, "QuorumAckChecker");

                //第三步：清理 pendingAppendResponsesByTerm、peerWaterMarksByTerm 中 是本次投票轮次的数据，避免一些不必要的内存使用
                if (pendingAppendResponsesByTerm.size() > 1) {
                    for (Long term : pendingAppendResponsesByTerm.keySet()) {
                        //假如是 当前轮次的直接跳过
                        if (term == currTerm) {
                            continue;
                        }
                        //不是当前轮次的
                        //<peerId, future> 执行结束操作 说明什么???
                        for (Map.Entry<Long, TimeoutFuture<AppendEntryResponse>> futureEntry : pendingAppendResponsesByTerm.get(term).entrySet()) {
                            AppendEntryResponse response = new AppendEntryResponse();
                            response.setGroup(memberState.getGroup());
                            response.setIndex(futureEntry.getKey());
                            response.setCode(DLedgerResponseCode.TERM_CHANGED.getCode());
                            response.setLeaderId(memberState.getLeaderId());
                            logger.info("[TermChange] Will clear the pending response index={} for term changed from {} to {}", futureEntry.getKey(), term, currTerm);
                            //结束调节
                            futureEntry.getValue().complete(response);
                        }//for over
                        //接着 又移除 本轮次的数据
                        pendingAppendResponsesByTerm.remove(term);
                    }//for over
                }

                //第四步：清除已过期的日志复制水位线，即投票轮次 不是 当前投票轮次的所有复制水位线都是过期数据，及时清除以避免内存泄露
                // 注意 :  这个数据结构是进行日志复制仲裁的关键，即主节点收到从节点的日志复制响应后，会将复制进度存储在该数据结构中
                if (peerWaterMarksByTerm.size() > 1) {
                    for (Long term : peerWaterMarksByTerm.keySet()) {
                        //当前轮次的 直接跳过
                        if (term == currTerm) {
                            continue;
                        }
                        logger.info("[TermChange] Will clear the watermarks for term changed from {} to {}", term, currTerm);
                        //否则直接删除
                        peerWaterMarksByTerm.remove(term);
                    }//for over
                }

                //peerWaterMarks 存储的是各个 从节点 当前已成功追加的日志序
                Map<String, Long> peerWaterMarks = peerWaterMarksByTerm.get(currTerm);
                //针对 日志序列号 进行排排序
                //todo 主节点在这里 不停的判断 并更新 主节点的提交 committedIndex
                //此处是 逆序 排列 ,  由于主节点 从一开始就是 水位线满状态 , 因此 逆序后 就是主节点满值是 第一个元素 ,  后面的就是 从节点
                List<Long> sortedWaterMarks = peerWaterMarks.values().stream().sorted(Comparator.reverseOrder()).collect(Collectors.toList());
                //System.out.println("sortedWaterMarks : " + sortedWaterMarks);
                //获得 数量/2 的节点索引 ->  也就是 从节点 成功响应后的 日志indexId
                long quorumIndex = sortedWaterMarks.get(sortedWaterMarks.size() / 2);
                final Optional<StateMachineCaller> fsmCaller = DLedgerEntryPusher.this.fsmCaller;
                //存在 这个 fsmCaller 究竟是啥 ???
                if (fsmCaller.isPresent()) {
                    // If there exist statemachine
                    //更新当前轮次的 已经确定的 quorumIndex  = committedIndex
                    DLedgerEntryPusher.this.dLedgerStore.updateCommittedIndex(currTerm, quorumIndex, "leader-fsmCaller");
                    //
                    final StateMachineCaller caller = fsmCaller.get();
                    //
                    caller.onCommitted(quorumIndex);

                    // Check elapsed
                    if (DLedgerUtils.elapsed(lastCheckLeakTimeMs) > 1000) {
                        //
                        updatePeerWaterMark(currTerm, memberState.getSelfId(), dLedgerStore.getLedgerEndIndex(), "fsmCaller.isPresent");
                        //
                        checkResponseFuturesElapsed(caller.getLastAppliedIndex());
                        //
                        lastCheckLeakTimeMs = System.currentTimeMillis();
                    }
                    //如果 当前
                    if (quorumIndex == this.lastQuorumIndex) {
                        //
                        waitForRunning(1);
                    }
                }
                //一般 走这个 不存在
                else {
                    // 更新 当前轮次的 索引
                    //System.out.println("quorumIndex : " + quorumIndex); //  -1  0  5  7 9
                    //todo 主节点在这里 不停的判断 并更新 主节点的提交 committedIndex  并且 主节点在向从节点发送数据时也会携带自己的提交指针
                    dLedgerStore.updateCommittedIndex(currTerm, quorumIndex, "leader-QuorumAckChecker-doWork()");
                    //第五步：根据各个节点的日志复制进度进行仲裁，确定已提交日志的序号，日志一旦提交，就可以向客户端返回写入成功
                    ConcurrentMap<Long, TimeoutFuture<AppendEntryResponse>> responses = pendingAppendResponsesByTerm.get(currTerm);
                    //是否需要检查 ???
                    boolean needCheck = false;
                    //回复客户端的 ack 数量
                    int ackNum = 0;
                    // 第六步：处理quorumIndex（已提交指针）之前的挂起请求 唤醒该日志序号之前挂起的请求，向客户端发送成功响应
                    // 从quorumIndex开始倒推，逐条向客户端返回写入成功请求
                    for (long i = quorumIndex; i > lastQuorumIndex; i--) {
                        try {
                            //从ConcurrentMap<Long,TimeoutFuture<AppendEntryResponse>>中移除对应的条目，并得到待通知的Future对象
                            CompletableFuture<AppendEntryResponse> future = responses.remove(i);
                            //如果从最后一个 都 未找到quorumIndex挂起请求，说明前面挂起的请求已经全部处理完毕，
                            // 则结束本次通知。结束之前再判断一次是否需要进行泄露检测
                            if (future == null) {
                                needCheck = true;
                                break;
                            } else if (!future.isDone()) {
                                AppendEntryResponse response = new AppendEntryResponse();
                                response.setGroup(memberState.getGroup());
                                response.setTerm(currTerm);
                                response.setIndex(i);
                                response.setLeaderId(memberState.getSelfId());
                                response.setPos(((AppendFuture<?>) future).getPos());
                                //调用future.complete方法向客户端返回响应结果
                                future.complete(response);
                            }
                            // ackNum加1，表示本次仲裁向客户端返回响应结果的数量
                            ackNum++;
                        } catch (Throwable t) {
                            logger.error("Error in ack to index={} term={}", i, currTerm, t);
                        }
                    }//for  quorumIndex--  over

                    //第七步：如果本次确认的个数为 0，则尝试去判断 超过该仲裁序号 的请求，是否已经超时，如果已超时，则返回超时响应结果
                    if (ackNum == 0) {
                        //判断 当前 已经提交过的后面的 日志请求 是否完成
                        checkResponseFuturesTimeout(quorumIndex + 1);
                        waitForRunning(1);
                    }

                    //第八步：进行日志追加挂起请求泄露检测，主要是遍历已挂起的请求，
                    // 如果日志序号小于已仲裁的序号，向客户端返回成功，将其移出待挂起队列
                    if (DLedgerUtils.elapsed(lastCheckLeakTimeMs) > 1000 || needCheck) {
                        //再去更新 主节点 - 水位点
                        updatePeerWaterMark(currTerm, memberState.getSelfId(), dLedgerStore.getLedgerEndIndex(), "needCheck");
                        //处理  小于 endIndex 之前的 客户端请求
                        checkResponseFuturesElapsed(quorumIndex);
                        lastCheckLeakTimeMs = System.currentTimeMillis();
                    }
                }
                //更新最新的 已提交索引
                lastQuorumIndex = quorumIndex;
            } catch (Throwable t) {
                DLedgerEntryPusher.logger.error("Error in {}", getName(), t);
                DLedgerUtils.sleep(100);
            }
        }
    }

    /**
     * This thread will be activated by the leader.
     * This thread will push the entry to follower(identified by peerId) and update the completed pushed index to index map.
     * Should generate a single thread for each peer.
     * The push has 4 types:
     * APPEND : append the entries to the follower
     * COMPARE : if the leader changes, the new leader should compare its entries to follower's
     * TRUNCATE : if the leader finished comparing by an index, the leader will send a request to truncate the follower's ledger
     * COMMIT: usually, the leader will attach the committed index with the APPEND request, but if the append requests are few and scattered,
     * the leader will send a pure request to inform the follower of committed index.
     * <p>
     * The common transferring between these types are as following:
     * <p>
     * COMPARE ---- TRUNCATE ---- APPEND ---- COMMIT
     * ^                             |
     * |---<-----<------<-------<----|
     * 日志转发线程，当前节点为主节点时 向从节点 转发数据
     * 主节点 为每一个 从节点构建一个 EntryDispatcher
     */
    private class EntryDispatcher extends ShutdownAbleThread {

        //向从节点发送命令的类型，可选值为 COMPARE、TRUNCATE、APPEND、COMMIT
        private AtomicReference<PushEntryRequest.Type> type = new AtomicReference<>(PushEntryRequest.Type.COMPARE);
        //上一次发送 commit 请求的时间戳
        private long lastPushCommitTimeMs = -1;
        //目标节点 ID  也就是 从节点 ID 唯一
        private String peerId;
        //已完成 compare 的日志序号
        private long compareIndex = -1;
        //writeIndex 表示当前已追加到从节点的 数据日志序号  不是 提交日志号
        private long writeIndex = -1;
        //允许的最大挂起日志数量
        private int maxPendingSize = 1000;
        //Leader节点当前的投票轮次。
        private long term = -1;
        //Leader节点ID
        private String leaderId = null;

        //上次检测泄漏的时间，所谓泄漏，指的是 挂起的日志请求数量是否超过了 maxPendingSize
        private long lastCheckLeakTimeMs = System.currentTimeMillis();

        //记录日志的挂起时间，key 表示日志的序列（entryIndex），value 表示挂起时间戳
        private ConcurrentMap<Long, Long> pendingMap = new ConcurrentHashMap<>();

        //批量日志的挂起时间
        private ConcurrentMap<Long, Pair<Long, Integer>> batchPendingMap = new ConcurrentHashMap<>();

        //转发实体
        private PushEntryRequest batchAppendEntryRequest = new PushEntryRequest();
        //配额
        private Quota quota = new Quota(dLedgerConfig.getPeerPushQuota());


        public EntryDispatcher(String peerId, Logger logger) {
            super("EntryDispatcher-" + memberState.getSelfId() + "-" + peerId, logger);
            this.peerId = peerId;
        }

        // checkAndFreshState()方法不只是简单地检测一下状态，
        // 而是会根据 状态机的运行状态 改变日志转发器的状态，
        // 从而驱动转发器是发送 append 请求还是发送 compare 请求
        private boolean checkAndFreshState() {
            //当前节点 不是 leader 了
            if (!memberState.isLeader()) {
                //直接返回
                return false;
            }
            // 如果日志转发器（EntryDispatcher）的 投票轮次为空 或 与状态机的投票轮次不相等，
            // 将日志转发器的 term、leaderId 与 状态 机 同步 , 即发送 compare 请求
            if (term != memberState.currTerm() || leaderId == null || !leaderId.equals(memberState.getLeaderId())) {
                //这种情况通常是由于集群触发了重新选举，当前节点刚被选举成 Leader 节点
                synchronized (memberState) {
                    if (!memberState.isLeader()) {
                        return false;
                    }
                    PreConditions.check(memberState.getSelfId().equals(memberState.getLeaderId()), DLedgerResponseCode.UNKNOWN);
                    term = memberState.currTerm();
                    leaderId = memberState.getSelfId();
                    //改变 日志转发器的状态
                    changeState(-1, PushEntryRequest.Type.COMPARE);
                }
            }
            return true;
        }

        private PushEntryRequest buildPushRequest(DLedgerEntry entry, PushEntryRequest.Type target) {
            PushEntryRequest request = new PushEntryRequest();
            request.setGroup(memberState.getGroup());
            request.setRemoteId(peerId);
            request.setLeaderId(leaderId);
            //默认是 -1
            request.setTerm(term);
            request.setEntry(entry);
            request.setType(target);
            //主节点向从节点发送数据时 会携带 本身已经 提交的序号 默认是 -1
            request.setCommitIndex(dLedgerStore.getCommittedIndex());
            return request;
        }

        private void resetBatchAppendEntryRequest() {
            batchAppendEntryRequest.setGroup(memberState.getGroup());
            batchAppendEntryRequest.setRemoteId(peerId);
            batchAppendEntryRequest.setLeaderId(leaderId);
            batchAppendEntryRequest.setTerm(term);
            batchAppendEntryRequest.setType(PushEntryRequest.Type.APPEND);
            batchAppendEntryRequest.clear();
        }

        private void checkQuotaAndWait(DLedgerEntry entry) {
            //1）append 挂起请求数已超过最大允许挂起数，默认为1000
            if (dLedgerStore.getLedgerEndIndex() - entry.getIndex() <= maxPendingSize) {
                return;
            }
            if (dLedgerStore instanceof DLedgerMemoryStore) {
                return;
            }
            DLedgerMmapFileStore mmapFileStore = (DLedgerMmapFileStore) dLedgerStore;
            //2）主从同步差异超过300MB，可通过peerPushThrottlePoint进行配置
            if (mmapFileStore.getDataFileList().getMaxWrotePosition() - entry.getPos() < dLedgerConfig.getPeerPushThrottlePoint()) {
                return;
            }
            //3）每秒追加的日志超过20MB（可通过peerPushQuota进行配置），则会休眠1s中后再追加
            quota.sample(entry.getSize());
            //
            if (quota.validateNow()) {
                long leftNow = quota.leftNow();
                logger.warn("[Push-{}]Quota exhaust, will sleep {}ms", peerId, leftNow);
                DLedgerUtils.sleep(leftNow);
            }
        }

        //EntryDispatcher 类 主节点使用 转发给从节点  日志
        private void doAppendInner(long index) throws Exception {
            //主节点 首先根据日志序号从 pagecache 中查询出日志数据实体
            DLedgerEntry entry = getDLedgerEntryForAppend(index);
            if (null == entry) {
                return;
            }
            //第六步：根据日志序号查询对应的日志内容，并检查是否超出配额，如果超出会触发限流，触发规则如下
            checkQuotaAndWait(entry);
            //第七步：构建append请求包，此时 请求中 携带 主节点的提交指针 , 通过网络异步发送并返回Future，
            //然后保存该条日志的发送时间戳，用于区分推送请求是否发送超时 , 从而触发重推
            PushEntryRequest request = buildPushRequest(entry, PushEntryRequest.Type.APPEND);
            // System.out.println("push 之前 " + request); // 0
            //todo 网络通信 注意 : 此时没有更新从节点的水位线
            System.out.println("doAppendInner() : " + request);
            CompletableFuture<PushEntryResponse> responseFuture = dLedgerRpcService.push(request);
            //重新设置新时间戳
            pendingMap.put(index, System.currentTimeMillis());
            //第八步：Leader节点收到从节点的append响应结果后，对结果进行处理，其核心逻辑如下
            responseFuture.whenComplete((x, ex) -> {
                try {
                    PreConditions.check(ex == null, DLedgerResponseCode.UNKNOWN);
                    DLedgerResponseCode responseCode = DLedgerResponseCode.valueOf(x.getCode());
                    switch (responseCode) {
                        //1. 移除pendingMap中对应的日志条目，表示已经成功收到响应结果
                        //2. 更新已成功追加的日志序号（按投票轮次组织，并且每个从服务器一个键值对）
                        //3. 唤醒quorumAckChecker线程（主要用于仲裁append结果）
                        case SUCCESS:
                            //成功了  移除负载队列
                            pendingMap.remove(x.getIndex());
                            //todo 更新 当前轮次 对方- 从节点的日志 水位线  -> 目的:  [QuorumAckChecker.doWork()仲裁器循环判断]
                            updatePeerWaterMark(x.getTerm(), peerId, x.getIndex(), "doAppendInner");
                            //日志仲裁器 唤醒
                            quorumAckChecker.wakeup();
                            break;
                        //如果append请求出现状态不一致的情况，Leader节点将发送
                        //compare请求到从节点，以便对比主从节点的数据是否一致
                        case INCONSISTENT_STATE:
                            logger.info("[Push-{}]Get INCONSISTENT_STATE when push index={} term={}", peerId, x.getIndex(), x.getTerm());
                            changeState(-1, PushEntryRequest.Type.COMPARE);
                            break;
                        default:
                            logger.warn("[Push-{}] Get error response code {} {}", peerId, responseCode, x.baseInfo());
                            break;
                    }
                } catch (Throwable t) {
                    logger.error("", t);
                }
            });
            lastPushCommitTimeMs = System.currentTimeMillis();
        }

        private DLedgerEntry getDLedgerEntryForAppend(long index) {
            DLedgerEntry entry;
            try {
                entry = dLedgerStore.get(index);
            } catch (DLedgerException e) {
                //  Do compare, in case the ledgerBeginIndex get refreshed.
                if (DLedgerResponseCode.INDEX_LESS_THAN_LOCAL_BEGIN.equals(e.getCode())) {
                    logger.info("[Push-{}]Get INDEX_LESS_THAN_LOCAL_BEGIN when requested index is {}, try to compare", peerId, index);
                    changeState(-1, PushEntryRequest.Type.COMPARE);
                    return null;
                }
                throw e;
            }
            PreConditions.check(entry != null, DLedgerResponseCode.UNKNOWN, "writeIndex=%d", index);
            return entry;
        }

        private void doCommit() throws Exception {
            if (DLedgerUtils.elapsed(lastPushCommitTimeMs) > 1000) {
                //从节点 发送 commit 指令
                PushEntryRequest request = buildPushRequest(null, PushEntryRequest.Type.COMMIT);
                System.out.println("doCommit() : " + request);
                //Ignore the results
                dLedgerRpcService.push(request);
                lastPushCommitTimeMs = System.currentTimeMillis();
            }
        }

        //检查并追加请求
        //该方法的作用是检查 append 请求是否超时
        private void doCheckAppendResponse() throws Exception {
            // 获取已成功 append() 的 日志序号
            long peerWaterMark = getPeerWaterMark(term, peerId);
            // 从挂起的请求队列中获取下一条的发送时间，如果不为空并去超过了append 的超时时间，
            // 则再重新发送 append 请求，最大超时时间默认为 1s，
            // 可以通过maxPushTimeOutMs 来改变默认值
            Long sendTimeMs = pendingMap.get(peerWaterMark + 1);
            //还在有效期
            if (sendTimeMs != null && System.currentTimeMillis() - sendTimeMs > dLedgerConfig.getMaxPushTimeOutMs()) {
                logger.warn("[Push-{}]Retry to push entry at {}", peerId, peerWaterMark + 1);
                //追加请求
                doAppendInner(peerWaterMark + 1);
            }
        }

        //目的 : Leader节点在确认主从数据一致后，开始将新的消息转发到从节点
        private void doAppend() throws Exception {
            //死循环
            while (true) {
                //第一步：再次判断节点状态，确保当前节点是Leader节点并且日志转发器内部的状态为 APPEND
                if (!checkAndFreshState()) {
                    break;
                }
                //
                if (type.get() != PushEntryRequest.Type.APPEND) {
                    break;
                }
                //第二步：writeIndex 表示当前已追加到 从节点的日志序号。通常情况下，
                // 主节点向从节点发送append请求时会带上主节点已提交的指针，
                // 但如果append请求发送不频繁，pending 请求超过其队列长度（默认为1万字节）时，
                // 会阻止数据的追加，此时有可能会出现 writeIndex 大于 leaderEndIndex 的情况，
                // 需要单独发送commit请求，并检查append请求响应
                // LedgerEndIndex 更新 不及时
                if (writeIndex > dLedgerStore.getLedgerEndIndex()) {
                    //
                    doCommit();
                    //
                    doCheckAppendResponse();
                    break;
                }
                //第三步：检测pendingMap（挂起的请求数量）是否发生泄露，
                // 泄漏 : 即挂起队列的容量是否超过了最大挂起阈值。获取当前节点的水位线（已成功append请求的日志序号），
                // 如果挂起请求的日志序号小于水位线，则丢弃，并记录最后一次检查的时间戳
                if (pendingMap.size() >= maxPendingSize || (DLedgerUtils.elapsed(lastCheckLeakTimeMs) > 1000)) {
                    //日志序列号
                    long peerWaterMark = getPeerWaterMark(term, peerId);
                    for (Long index : pendingMap.keySet()) {
                        //如果挂起请求的日志序号小于 水位线， 说明 已经 转发完毕了 , 则丢弃
                        if (index < peerWaterMark) {
                            pendingMap.remove(index);
                        }
                    }
                    lastCheckLeakTimeMs = System.currentTimeMillis();
                }

                //第四步：如果挂起的请求（等待从节点追加结果）大于 maxPendingSize ，
                //并且下一条日志的推送请求已经超时，则重新发起推送请求，即重新发送日志，避免网络丢包等异常
                if (pendingMap.size() >= maxPendingSize) {
                    //继续检查针对当前peerId的 pendingMap 队列中的值
                    doCheckAppendResponse();
                    //不执行转发操作 并 退出
                    break;
                }
                //第五步：调用 doAppendInner() 方法将日志转发到从节点
                doAppendInner(writeIndex);
                //转发数量 增加
                writeIndex++;
            }//while over
        }

        private void sendBatchAppendEntryRequest() throws Exception {
            batchAppendEntryRequest.setCommitIndex(dLedgerStore.getCommittedIndex());
            CompletableFuture<PushEntryResponse> responseFuture = dLedgerRpcService.push(batchAppendEntryRequest);
            batchPendingMap.put(batchAppendEntryRequest.getFirstEntryIndex(), new Pair<>(System.currentTimeMillis(), batchAppendEntryRequest.getCount()));
            responseFuture.whenComplete((x, ex) -> {
                try {
                    PreConditions.check(ex == null, DLedgerResponseCode.UNKNOWN);
                    DLedgerResponseCode responseCode = DLedgerResponseCode.valueOf(x.getCode());
                    switch (responseCode) {
                        case SUCCESS:
                            batchPendingMap.remove(x.getIndex());
                            updatePeerWaterMark(x.getTerm(), peerId, x.getIndex() + x.getCount() - 1, "sendBatchAppendEntryRequest");
                            break;
                        case INCONSISTENT_STATE:
                            logger.info("[Push-{}]Get INCONSISTENT_STATE when batch push index={} term={}", peerId, x.getIndex(), x.getTerm());
                            changeState(-1, PushEntryRequest.Type.COMPARE);
                            break;
                        default:
                            logger.warn("[Push-{}]Get error response code {} {}", peerId, responseCode, x.baseInfo());
                            break;
                    }
                } catch (Throwable t) {
                    logger.error("", t);
                }
            });
            lastPushCommitTimeMs = System.currentTimeMillis();
            batchAppendEntryRequest.clear();
        }

        private void doBatchAppendInner(long index) throws Exception {
            DLedgerEntry entry = getDLedgerEntryForAppend(index);
            if (null == entry) {
                return;
            }
            batchAppendEntryRequest.addEntry(entry);
            if (batchAppendEntryRequest.getTotalSize() >= dLedgerConfig.getMaxBatchPushSize()) {
                sendBatchAppendEntryRequest();
            }
        }

        //如果因网络等原因，主节点在向从节点追加日志时失败，
        // 该如何保证从节点与主节点一致呢？
        // 从上文我们可以得知，Leader节点在向从节点转发日志后，会存储该日志的推送时间戳到pendingMap，
        // 其存储结构为ConcurrentMap<Long/*日志序号*/, Long/*PUSH时间戳*/> pendingMap。
        // 当pendingMap的积压超过1000ms时会触发重推机制，该逻辑封装在doCheckAppendResponse()方法中
        private void doCheckBatchAppendResponse() throws Exception {
            //获取从节点已复制的日志序号 peerWaterMark
            long peerWaterMark = getPeerWaterMark(term, peerId);
            //然后用在pendingMap中尝试查找该日志序号加1的记录，
            // 如果能找到，说明从服务下一条需要追加的消息已经存储在主节点中，
            // 接着在尝试推送，如果该条推送已经超时，默认超时时间为1s，调用doAppendInner重新推送
            Pair pair = batchPendingMap.get(peerWaterMark + 1);
            //
            if (pair != null && System.currentTimeMillis() - (long) pair.getKey() > dLedgerConfig.getMaxPushTimeOutMs()) {
                //
                long firstIndex = peerWaterMark + 1;
                //
                long lastIndex = firstIndex + (int) pair.getValue() - 1;
                //
                logger.warn("[Push-{}]Retry to push entry from {} to {}", peerId, firstIndex, lastIndex);
                //
                batchAppendEntryRequest.clear();
                //
                for (long i = firstIndex; i <= lastIndex; i++) {
                    DLedgerEntry entry = dLedgerStore.get(i);
                    batchAppendEntryRequest.addEntry(entry);
                }
                //
                sendBatchAppendEntryRequest();
            }
        }

        private void doBatchAppend() throws Exception {
            while (true) {
                if (!checkAndFreshState()) {
                    break;
                }
                if (type.get() != PushEntryRequest.Type.APPEND) {
                    break;
                }
                if (writeIndex > dLedgerStore.getLedgerEndIndex()) {
                    if (batchAppendEntryRequest.getCount() > 0) {
                        sendBatchAppendEntryRequest();
                    }
                    doCommit();
                    doCheckBatchAppendResponse();
                    break;
                }
                if (batchPendingMap.size() >= maxPendingSize || (DLedgerUtils.elapsed(lastCheckLeakTimeMs) > 1000)) {
                    long peerWaterMark = getPeerWaterMark(term, peerId);
                    for (Map.Entry<Long, Pair<Long, Integer>> entry : batchPendingMap.entrySet()) {
                        if (entry.getKey() + entry.getValue().getValue() - 1 <= peerWaterMark) {
                            batchPendingMap.remove(entry.getKey());
                        }
                    }
                    lastCheckLeakTimeMs = System.currentTimeMillis();
                }
                if (batchPendingMap.size() >= maxPendingSize) {
                    doCheckBatchAppendResponse();
                    break;
                }
                doBatchAppendInner(writeIndex);
                writeIndex++;
            }
        }

        // Leader节点在发送compare() 请求后，得知与从节点的数据存在差异，
        // 将向从节点发送 truncate 请求，指示从节点应该将 truncateIndex 及以后的日志删除
        private void doTruncate(long truncateIndex) throws Exception {
            // 该方法的实现比较简单，主节点构建truncate请求包并通过网络向从节点发送请求，
            // 从节点在收到请求后会清理多余的数据，使主从节点数据保持一致。
            // 日志转发器在处理完truncate请求后，状态将变更为APPEND，开始向从节点转发日志
            PreConditions.check(type.get() == PushEntryRequest.Type.TRUNCATE, DLedgerResponseCode.UNKNOWN);
            //
            DLedgerEntry truncateEntry = dLedgerStore.get(truncateIndex);
            //
            PreConditions.check(truncateEntry != null, DLedgerResponseCode.UNKNOWN);
            //
            logger.info("[Master Push to the - {} ]will be push data to truncate truncateIndex={} pos={}", peerId, truncateIndex, truncateEntry.getPos());
            //
            PushEntryRequest truncateRequest = buildPushRequest(truncateEntry, PushEntryRequest.Type.TRUNCATE);
            System.out.println("doTruncate() : " + truncateRequest);
            //todo 网络 push
            PushEntryResponse truncateResponse = dLedgerRpcService.push(truncateRequest).get(3, TimeUnit.SECONDS);
            //
            PreConditions.check(truncateResponse != null, DLedgerResponseCode.UNKNOWN, "truncateIndex=%d", truncateIndex);
            //这把得必须 成功 SUCCESS
            PreConditions.check(truncateResponse.getCode() == DLedgerResponseCode.SUCCESS.getCode(), DLedgerResponseCode.valueOf(truncateResponse.getCode()), "truncateIndex=%d", truncateIndex);
            //
            lastPushCommitTimeMs = System.currentTimeMillis();
            //
            changeState(truncateIndex, PushEntryRequest.Type.APPEND);
        }

        // 改变日志转发器的状态
        //1）Long index：已写入日志序号
        //2）PushEntryRequest.Type target：主节点 日志转发器 即将进入的状态
        private synchronized void changeState(long index, PushEntryRequest.Type target) {
            logger.info("[Push-{}]Change state from {} to {} at {}", peerId, type.get(), target, index);
            switch (target) {
                // 日志转发器即将进入APPEND状态，该状态下主节点将向从节点转发日志，
                // 重置compareIndex指针，更新当前节点已追加日志序号为index，
                // 并唤醒QuorumAckChecker线程，以便对append响应结果进行仲裁。
                // 最后更新待追加日志序号（writeIndex）
                case APPEND:
                    compareIndex = -1;
                    //更新目标 Id 的 水位
                    updatePeerWaterMark(term, peerId, index, "949 changeState " + target.name());
                    //唤醒仲裁线程
                    quorumAckChecker.wakeup();
                    //更新待追加日志序号（writeIndex）
                    writeIndex = index + 1;
                    if (dLedgerConfig.isEnableBatchPush()) {
                        resetBatchAppendEntryRequest();
                    }
                    break;
                //日志转发器即将进入COMPARE状态，该状态下先重置compareIndex为-1，
                // 然后清除已挂起的日志转发请求，与从节点进行协商，以确保主从节点数据一致
                case COMPARE:
                    if (this.type.compareAndSet(PushEntryRequest.Type.APPEND, PushEntryRequest.Type.COMPARE)) {
                        //先重置compareIndex为-1
                        compareIndex = -1;
                        if (dLedgerConfig.isEnableBatchPush()) {
                            //清除挂起队列
                            batchPendingMap.clear();
                        } else {
                            //清除挂起队列
                            pendingMap.clear();
                        }
                    }
                    break;
                //日志转发器即将进入TRUNCATE状态，该状态下先重置compareIndex为-1，
                // 然后向从节点发起truncate请求，清除从节点未提交且在主节点上不存在的数据，
                // 确保主从节点数据一致
                case TRUNCATE:
                    compareIndex = -1;
                    break;
                default:
                    break;
            }
            //type 就直接 set 了
            type.set(target);
        }

        //日志转发器EntryDispatcher的初始状态为COMPARE，当一个节点
        //被选举为Leader后，日志转发器的状态同样会先设置为COMPARE，
        //Leader节点先向从节点发送该请求的目的是比较主、
        // 从节点之间数据的差异，以此确保发送主从切换时不会丢失数据，并且重新确定待转发的日志序号
        private void doCompare() throws Exception {
            while (true) {
                //第一步：再次验证当前状态下是否可以发送compare请求
                if (!checkAndFreshState()) {
                    break;
                }
                //如果请求类型不是 compare或truncate，则直接跳出
                if (type.get() != PushEntryRequest.Type.COMPARE && type.get() != PushEntryRequest.Type.TRUNCATE) {
                    break;
                }
                //如果compareIndex和ledgerEndIndex都为-1，表示这是一个新的集群，没有存储任何数据，故无须比较主从是否一致
                if (compareIndex == -1 && dLedgerStore.getLedgerEndIndex() == -1) {
                    break;
                }

                // revise the compareIndex &&  dLedgerStore.getLedgerEndIndex() != -1
                //第二步：重置 compareIndex，其实现逻辑为如果compareIndex为-1或compareIndex不在有效范围内，
                // 则重置 compareIndex 为 Leader 节点当前存储的最大日志序号
                if (compareIndex == -1) {
                    compareIndex = dLedgerStore.getLedgerEndIndex();
                    logger.info("[Push-{}][DoCompare] compareIndex=-1 means start to compare", peerId);
                }
                //
                else if (compareIndex > dLedgerStore.getLedgerEndIndex() || compareIndex < dLedgerStore.getLedgerBeginIndex()) {
                    logger.info("[Push-{}][DoCompare] compareIndex={} out of range {}-{}", peerId, compareIndex, dLedgerStore.getLedgerBeginIndex(), dLedgerStore.getLedgerEndIndex());
                    compareIndex = dLedgerStore.getLedgerEndIndex();
                }
                //获得实体
                DLedgerEntry entry = dLedgerStore.get(compareIndex);
                PreConditions.check(entry != null, DLedgerResponseCode.INTERNAL_ERROR, "compareIndex=%d", compareIndex);
                //构建请求
                PushEntryRequest request = buildPushRequest(entry, PushEntryRequest.Type.COMPARE);
                System.out.println("doCompare() : request " + request);
                //第三步：根据待比较的日志序号查询日志，并向 从节点发起 compare 请求，默认超时时间为3s
                CompletableFuture<PushEntryResponse> responseFuture = dLedgerRpcService.push(request);
                PushEntryResponse response = responseFuture.get(3, TimeUnit.SECONDS);
                PreConditions.check(response != null, DLedgerResponseCode.INTERNAL_ERROR, "compareIndex=%d", compareIndex);
                PreConditions.check(response.getCode() == DLedgerResponseCode.INCONSISTENT_STATE.getCode() || response.getCode() == DLedgerResponseCode.SUCCESS.getCode(), DLedgerResponseCode.valueOf(response.getCode()), "compareIndex=%d", compareIndex);
                //第四步：根据从节点的响应结果计算 truncateIndex（需要截断的日志序号），即计算 从节点中多余的数据
                long truncateIndex = -1;

                if (response.getCode() == DLedgerResponseCode.SUCCESS.getCode()) {
                    System.out.println("doCompare() :  " + response);
                    /*
                     * The comparison is successful:
                     * 1. Just change to append state, if the follower's end index is equal the compared index.
                     * 2. Truncate the follower, if the follower has some dirty entries.
                     */
                    //如果从节点已存储的日志序号与主节点比较过的相同，那么就无须截断，
                    // 日志转发器的状态将变更为 APPEND，主节点将开始向从节点转发日志，
                    if (compareIndex == response.getEndIndex()) {
                        //流程比较
                        changeState(compareIndex, PushEntryRequest.Type.APPEND);
                        //直接返回
                        break;
                    }
                    // 那么就说明不一样 需要 设置 truncateIndex 为从节点返回的 compareIndex，将向从节点发送 truncate 请求
                    else {
                        truncateIndex = compareIndex;
                    }
                }
                //如果从节点存储的最大日志序号小于主节点的最小序号，或者
                //从节点的最小日志序号大于主节点的最大日志序号，即两者不相交，这通常发生在从节点崩溃很长一段时间，而主节点删除了过期的条目时 ,
                //则设置truncateIndex为主节点的ledgerBeginIndex，即主节点目前最小的偏移量，
                // 这样意味着将会删除从节点所有的数据，
                // 然后从truncateIndex开始向从节点重新转发日志
                else if (response.getEndIndex() < dLedgerStore.getLedgerBeginIndex() || response.getBeginIndex() > dLedgerStore.getLedgerEndIndex()) {
                    /*
                     The follower's entries does not intersect with the leader.
                     This usually happened when the follower has crashed for a long time while the leader has deleted the expired entries.
                     Just truncate the follower.
                     */
                    truncateIndex = dLedgerStore.getLedgerBeginIndex();
                }
                //如果 compareIndex 小于 从节点 的开始日志序号，则从主节点最小日志序号开始同步
                else if (compareIndex < response.getBeginIndex()) {
                    /*
                     The compared index is smaller than the follower's begin index.
                     This happened rarely, usually means some disk damage.
                     Just truncate the follower.
                     */
                    truncateIndex = dLedgerStore.getLedgerBeginIndex();
                }
                //如果 compareIndex 大于从节点的最大日志序号，并且不小于 Leader 节点的最小存储日志序号，
                //则将 compareIndex 设置为从节点最大的日志序号，继续发起 compare 请求
                else if (compareIndex > response.getEndIndex()) {
                    /*
                     The compared index is bigger than the follower's end index.
                     This happened frequently. For the compared index is usually starting from the end index of the leader.
                     */
                    compareIndex = response.getEndIndex();
                }
                //如果 compareIndex 大于从节点的开始日志序号，但小于从节点的最大日志序号，
                //  response.getBeginIndex()< compareIndex < response.getEndIndex()
                // 表示主节点与从节点数据有相交的情况，故将 compareIndex 减1，继续比较，直到找到需要截断的日志序号
                else {
                    /*
                      Compare failed and the compared index is in the range of follower's entries.
                     */
                    compareIndex--;
                }

                /*`
                 * The compared index is smaller than the leader's begin index, truncate the follower.
                 */
                //如果 compareIndex 小于主节点的最小日志序号，则将 truncateIndex 设置为主节点的最小日志序号
                if (compareIndex < dLedgerStore.getLedgerBeginIndex()) {
                    truncateIndex = dLedgerStore.getLedgerBeginIndex();
                }

                /*
                 If get value for truncateIndex, do it right now.
                 */
                //第五步：如果 truncateIndex 不等于-1，则日志转发器状态设置为 TRUNCATE，
                // 然后向从节点发送truncate请求，具体由 doTruncate() 方法实现
                if (truncateIndex != -1) {
                    changeState(truncateIndex, PushEntryRequest.Type.TRUNCATE);
                    doTruncate(truncateIndex);
                    break;
                }

            }//while over
        }

        @Override
        public void doWork() {
            try {
                //检查当前节点的状态，确定当前节点状态是否可以发送append或compare请求
                //checkAndFreshState() 方法不只是简单地检测一下状态，而是会
                //根据运行状态改变日志转发器的状态，从而驱动转发器是发送append请求还是发送compare请求
                if (!checkAndFreshState()) {
                    waitForRunning(1);
                    return;
                }

                //根据当前转发器的状态向从节点发送 append 或 compare 请求
                //转发器 默认是  COMPARE 状态 也就是说明 需要先跟从节点进行比较
                if (type.get() == PushEntryRequest.Type.APPEND) {
                    //是否 开启批量 转发  默认是 false
                    if (dLedgerConfig.isEnableBatchPush()) {
                        doBatchAppend();
                    } else {
                        //
                        doAppend();
                    }
                }
                // 日志转发器 EntryDispatcher 的初始状态为 COMPARE，当一个节点被选举为Leader后，
                // 日志转发器的状态同样会先设置为 COMPARE，
                // Leader 节点 先向 从节点发送该请求的目的是比较主、从节点之间数据的差异，以此确保发送主从切换时不会丢失数据，
                // 并且重新确定待转发的日志序号
                else {
                    //其余的命令就是  COMMIT || COMPARE || TRUNCATE
                    doCompare();
                }
                //阻塞等着
                waitForRunning(1);
            } catch (Throwable t) {
                DLedgerEntryPusher.logger.error("[Push-{}]Error in {} writeIndex={} compareIndex={}", peerId, getName(), writeIndex, compareIndex, t);
                changeState(-1, PushEntryRequest.Type.COMPARE);
                DLedgerUtils.sleep(500);
            }
        }
    }

    /**
     * This thread will be activated by the follower.
     * Accept the push request and order it by the index, then append to ledger store one by one.
     * 从节点开启
     */
    private class EntryHandler extends ShutdownAbleThread {

        //上一次检查主服务器是否有推送消息的时间戳
        private long lastCheckFastForwardTimeMs = System.currentTimeMillis();

        //append() 请求处理队列 这里为何 没有使用其他具有排序性质的队列??? 写多 删多 情况
        //<index , entryRequest ,future>>
        ConcurrentMap<Long, Pair<PushEntryRequest, CompletableFuture<PushEntryResponse>>> writeRequestMap = new ConcurrentHashMap<>();

        //COMMIT、COMPARE、TRUNCATE 相关请求的处理队列
        BlockingQueue<Pair<PushEntryRequest, CompletableFuture<PushEntryResponse>>> compareOrTruncateRequests = new ArrayBlockingQueue<Pair<PushEntryRequest, CompletableFuture<PushEntryResponse>>>(100);

        public EntryHandler(Logger logger) {
            super("EntryHandler-" + memberState.getSelfId(), logger);
        }

        //从节点收到Leader节点的推送请求后（无论是APPEND、COMMIT、COMPARE、TRUNCATE），由EntryHandler的handlePush()方法执行
        //目的 : handlePush()方法的主要职责是将处理请求放入队列，由doWork()方法从处理队列中拉取任务进行处理
        public CompletableFuture<PushEntryResponse> handlePush(PushEntryRequest request) throws Exception {
            //System.out.println("handlePush : " + request);  // commitIndex :  -1  0  0  0 9
            //The timeout should smaller than the remoting layer's request timeout
            //先构建 请求 响应对应的超时 future
            CompletableFuture<PushEntryResponse> future = new TimeoutFuture<>(1000);
            //再次 区分 消息 类型
            switch (request.getType()) {
                //1）如果是append请求，将请求放入writeRequestMap集合，如果
                //已存在该条日志的推送请求，表示Leader重复推送，则返回状态码 REPEATED_PUSH
                case APPEND:
                    if (request.isBatch()) {
                        PreConditions.check(request.getBatchEntry() != null && request.getCount() > 0, DLedgerResponseCode.UNEXPECTED_ARGUMENT);
                    } else {
                        PreConditions.check(request.getEntry() != null, DLedgerResponseCode.UNEXPECTED_ARGUMENT);
                    }
                    //消息日志 ID
                    long index = request.getFirstEntryIndex();
                    //放入 从节点的待追加队列中 [写操作]
                    Pair<PushEntryRequest, CompletableFuture<PushEntryResponse>> old = writeRequestMap.putIfAbsent(index, new Pair<>(request, future));
                    //存在重复的 说明主节点重复推送
                    if (old != null) {
                        logger.warn("[MONITOR]The index {} has already existed with {} and curr is {}", index, old.getKey().baseInfo(), request.baseInfo());
                        //返回客户端响应   REPEATED_PUSH
                        future.complete(buildResponse(request, DLedgerResponseCode.REPEATED_PUSH.getCode()));
                    }
                    break;
                //2）如果是commit请求，将请求存入 compareOrTruncateRequests
                //请求处理队列
                case COMMIT:
                    compareOrTruncateRequests.put(new Pair<>(request, future));
                    break;
                //3）如果是compare或truncate请求，将待追加队列
                //writeRequestMap清空，并将请求放入compareOrTruncateRequests请
                //求队列，由doWork()方法进行异步处理
                case COMPARE://纠错
                case TRUNCATE://拦截
                    PreConditions.check(request.getEntry() != null, DLedgerResponseCode.UNEXPECTED_ARGUMENT);
                    //清空 一切重来
                    writeRequestMap.clear();
                    compareOrTruncateRequests.put(new Pair<>(request, future));
                    break;
                default:
                    logger.error("[BUG]Unknown type {} from {}", request.getType(), request.baseInfo());
                    future.complete(buildResponse(request, DLedgerResponseCode.UNEXPECTED_ARGUMENT.getCode()));
                    break;
            }
            //唤醒  doWork() 处线程 操作
            wakeup();
            //仅仅是返回 没有明确的返回客户端响应 并没有执行 complete()  因为是在 dLedgerStore.appendAsFollower() 执行的
            return future;
        }

        //主要也是返回当前从节点的 ledgerBeginIndex、ledgerEndIndex 以及投票轮次，供主节点进行判断比较
        private PushEntryResponse buildResponse(PushEntryRequest request, int code) {
            PushEntryResponse response = new PushEntryResponse();
            response.setGroup(request.getGroup());
            response.setCode(code);
            //当前从节点的轮次
            response.setTerm(request.getTerm());
            //APPEND || COMPARE || TRUNCATE 请求
            if (request.getType() != PushEntryRequest.Type.COMMIT) {
                //Leader 下发 不合理的第一个消息 indexId
                response.setIndex(request.getFirstEntryIndex());
                //消息数量
                response.setCount(request.getCount());
            }
            //同时将当前从节点已存储的 最小日志序号、最大日志序号、当前投票轮次返回给Leader节点，
            //方便Leader节点进行比较，从而计算出应该截断的日志序号
            response.setBeginIndex(dLedgerStore.getLedgerBeginIndex());
            response.setEndIndex(dLedgerStore.getLedgerEndIndex());
            return response;
        }

        //从节点 调用 DLedgerStore 的 appendAsFollower() 方法进行日志的追加，与 appendAsLeader() 在日志存储部分相同，只是从节点无需再转发日志
        private void handleDoAppend(long writeIndex, PushEntryRequest request, CompletableFuture<PushEntryResponse> future) {
            try {
                PreConditions.check(writeIndex == request.getEntry().getIndex(), DLedgerResponseCode.INCONSISTENT_STATE);
                //与 主节点 收到客户端的写请求时 一样的操作处理流程 只是放在了从节点的 pagecache 中
                System.out.println("handleDoAppend() request : " + request);
                DLedgerEntry entry = dLedgerStore.appendAsFollower(request.getEntry(), request.getTerm(), request.getLeaderId());
                PreConditions.check(entry.getIndex() == writeIndex, DLedgerResponseCode.INCONSISTENT_STATE);
                //System.out.println("handleDoAppend : " + request); request.getCommitIndex() == 0
                PushEntryResponse response = buildResponse(request, DLedgerResponseCode.SUCCESS.getCode());
                System.out.println("handleDoAppend() response : " + response);
                future.complete(response);
                //提交从节点的提交位移  request.getCommitIndex()
                updateCommittedIndex(request.getTerm(), request.getCommitIndex(), "follower-handleDoAppend()");
            } catch (Throwable t) {
                logger.error("[HandleDoWrite] writeIndex={}", writeIndex, t);
                future.complete(buildResponse(request, DLedgerResponseCode.INCONSISTENT_STATE.getCode()));
            }
        }

        //处理主节点发送过来的 COMPARE 请求，其实现也比较简单，最终调用 buildResponse()  方法构造响应结果
        private CompletableFuture<PushEntryResponse> handleDoCompare(long compareIndex, PushEntryRequest request, CompletableFuture<PushEntryResponse> future) {
            try {
                PreConditions.check(compareIndex == request.getEntry().getIndex(), DLedgerResponseCode.UNKNOWN);
                PreConditions.check(request.getType() == PushEntryRequest.Type.COMPARE, DLedgerResponseCode.UNKNOWN);
                System.out.println("handleDoCompare() request : " + request);
                //核心思想是判断 Leader 节点传来的 日志序号在从节点中是否存在
                DLedgerEntry local = dLedgerStore.get(compareIndex);
                System.out.println("handleDoCompare() local : " + local);
                //不相等会抛异常
                PreConditions.check(request.getEntry().equals(local), DLedgerResponseCode.INCONSISTENT_STATE);
                //没有抛出异常则 正常返回
                PushEntryResponse pushEntryResponse = buildResponse(request, DLedgerResponseCode.SUCCESS.getCode());
                System.out.println("handleDoCompare() Response : " + pushEntryResponse);
                future.complete(pushEntryResponse);
            } catch (Throwable t) {
                logger.error("[HandleDoCompare] compareIndex={}", compareIndex, t);
                //捕获异常 发送 INCONSISTENT_STATE 响应
                PushEntryResponse response = buildResponse(request, DLedgerResponseCode.INCONSISTENT_STATE.getCode());
                System.out.println("handleDoCompare() err Response : " + response);
                future.complete(response);
            }
            return future;
        }

        //处理提交请求，其处理比较简单，就是调用 DLedgerStore 的updateCommittedIndex 更新其已提交偏移量，
        // 故我们还是具体看一下 DLedgerStore 的updateCommittedIndex 方法
        private CompletableFuture<PushEntryResponse> handleDoCommit(long committedIndex, PushEntryRequest request, CompletableFuture<PushEntryResponse> future) {
            try {
                PreConditions.check(committedIndex == request.getCommitIndex(), DLedgerResponseCode.UNKNOWN);
                PreConditions.check(request.getType() == PushEntryRequest.Type.COMMIT, DLedgerResponseCode.UNKNOWN);
                System.out.println("handleDoCommit() request : " + request);
                //重点
                updateCommittedIndex(request.getTerm(), committedIndex, "follower-handleDoCommit()");
                PushEntryResponse response = buildResponse(request, DLedgerResponseCode.SUCCESS.getCode());
                System.out.println("handleDoCommit() response :" + response);
                future.complete(response);
            } catch (Throwable t) {
                logger.error("[HandleDoCommit] committedIndex={}", request.getCommitIndex(), t);
                future.complete(buildResponse(request, DLedgerResponseCode.UNKNOWN.getCode()));
            }
            return future;
        }

        // Leader节点 与 从节点进行数据对比后，如果发现数据有差异，
        // 将计算出需要 截断的日志序号，发送 truncate() 请求给从节点，从节点对多余的日志进行 截断
        private CompletableFuture<PushEntryResponse> handleDoTruncate(long truncateIndex, PushEntryRequest request, CompletableFuture<PushEntryResponse> future) {
            try {
                logger.info("[HandleDoTruncate] truncateIndex={} pos={}", truncateIndex, request.getEntry().getPos());
                System.out.println("handleDoTruncate() request : " + request);
                //应该相同
                PreConditions.check(truncateIndex == request.getEntry().getIndex(), DLedgerResponseCode.UNKNOWN);
                PreConditions.check(request.getType() == PushEntryRequest.Type.TRUNCATE, DLedgerResponseCode.UNKNOWN);
                //已删除的 日志Id
                long index = dLedgerStore.truncate(request.getEntry(), request.getTerm(), request.getLeaderId());
                // 删除节点上truncateIndex日志序号之后的所有日志，会调用dLedgerStore的truncate()方法，
                // 根据日志序号定位到日志文件。如果命中具体的文件，则修改相应的读写指针、刷盘指针等，并将所在物理文件之后的所有文件删除
                PreConditions.check(index == truncateIndex, DLedgerResponseCode.INCONSISTENT_STATE);
                PushEntryResponse response = buildResponse(request, DLedgerResponseCode.SUCCESS.getCode());
                System.out.println("handleDoTruncate() response : " + response);
                future.complete(response);
                updateCommittedIndex(request.getTerm(), request.getCommitIndex(), "follower-handleDoTruncate()");
            } catch (Throwable t) {
                logger.error("[HandleDoTruncate] truncateIndex={}", truncateIndex, t);
                future.complete(buildResponse(request, DLedgerResponseCode.INCONSISTENT_STATE.getCode()));
            }
            return future;
        }

        private void handleDoBatchAppend(long writeIndex, PushEntryRequest request, CompletableFuture<PushEntryResponse> future) {
            try {
                PreConditions.check(writeIndex == request.getFirstEntryIndex(), DLedgerResponseCode.INCONSISTENT_STATE);
                for (DLedgerEntry entry : request.getBatchEntry()) {
                    dLedgerStore.appendAsFollower(entry, request.getTerm(), request.getLeaderId());
                }
                future.complete(buildResponse(request, DLedgerResponseCode.SUCCESS.getCode()));
                updateCommittedIndex(request.getTerm(), request.getCommitIndex(), "follower-handleDoBatchAppend()");
            } catch (Throwable t) {
                logger.error("[HandleDoBatchAppend]", t);
                future.complete(buildResponse(request, DLedgerResponseCode.INCONSISTENT_STATE.getCode()));
            }

        }

        // 收到Leader节点的append请求后，从节点首先会将这些写入请求存储在writeRequestMap()处理队列中，
        // 从节点并不是直接从该队列中获取一个待写入处理请求进行数据追加，
        // 而是查找当前节点已存储的 最大日志序号 leaderEndIndex，
        // 然后加1得出下一条待追加的日志序号 nextIndex。
        // 如果该日志序号在writeRequestMap中不存在日志推送请求，
        // 则有可能是因为发生了推送请求丢失，在这种情况下，需要进行异常检测，
        // 以便尽快恢复异常，使主节点与从节点最终保持一致性。
        //从节点的日志复制异常检测由checkAbnormalFuture()方法实现
        private void checkAppendFuture(long endIndex) {
            long minFastForwardIndex = Long.MAX_VALUE;
            //遍历所有待写请求 -> 此时 writeRequestMap 有可能为空  ||  也有能有数据了
            for (Pair<PushEntryRequest, CompletableFuture<PushEntryResponse>> pair : writeRequestMap.values()) {
                //获取追加队列 第一个
                long firstEntryIndex = pair.getKey().getFirstEntryIndex();
                //获取追加队列 最后一个
                long lastEntryIndex = pair.getKey().getLastEntryIndex();
                //Fall behind
                //如果待追加的最大日志序号 小于 从节点已经存储的最大日志序号  说明什么? 说明 理应丢弃的数据 需要再次检查一遍
                if (lastEntryIndex <= endIndex) {
                    try {
                        if (pair.getKey().isBatch()) {
                            //遍历 挂起队列中的数据的日志Id 看从节点中是否真的有了
                            for (DLedgerEntry dLedgerEntry : pair.getKey().getBatchEntry()) {
                                //从节点再从 自己本地查找出来进行对比 假如存在 不一致的信息 立马抛出 INCONSISTENT_STATE 异常返回主节点
                                PreConditions.check(dLedgerEntry.equals(dLedgerStore.get(dLedgerEntry.getIndex())), DLedgerResponseCode.INCONSISTENT_STATE);
                            }
                        } else {
                            DLedgerEntry dLedgerEntry = pair.getKey().getEntry();
                            //从节点再从 自己本地查找出来进行对比 假如存在 不一致的信息 立马抛出 INCONSISTENT_STATE 异常返回主节点
                            PreConditions.check(dLedgerEntry.equals(dLedgerStore.get(dLedgerEntry.getIndex())), DLedgerResponseCode.INCONSISTENT_STATE);
                        }
                        //不抛出异常 说明没得问题
                        pair.getValue().complete(buildResponse(pair.getKey(), DLedgerResponseCode.SUCCESS.getCode()));
                        logger.warn("[PushFallBehind]The leader pushed an batch append entry last index={} smaller than current ledgerEndIndex={}, maybe the last ack is missed", lastEntryIndex, endIndex);
                    } catch (Throwable t) {
                        logger.error("[PushFallBehind]The leader pushed an batch append entry last index={} smaller than current ledgerEndIndex={}, maybe the last ack is missed", lastEntryIndex, endIndex, t);
                        pair.getValue().complete(buildResponse(pair.getKey(), DLedgerResponseCode.INCONSISTENT_STATE.getCode()));
                    }
                    //没事了 就 并且执行删除 这一批 从 第一个开始 操作
                    writeRequestMap.remove(pair.getKey().getFirstEntryIndex());
                    //开始遍历下一个 pair
                    continue;
                }

                //第三步：如果待追加的日志序号 等于 endIndex+1，即 从节点当前存储的最大日志序号加1，
                // 表示从节点下一条期望追加的日志Leader节点已经推送过来了
                // 这种情况非常正常，故可结束异常检测逻辑
                if (firstEntryIndex == endIndex + 1) {
                    return;
                }

                // firstEntryIndex == endIndex || firstEntryIndex < endIndex : 可能会出现吗?
                // 当主节点发送的数据被从节点进行保存并返回ack时 , ack可能没有及时被主节点收到 , 就导致主节点的TCP进行网络重传
                // firstEntryIndex > endIndex
                // firstEntryIndex > endIndex +1
                // 第四步：主要 处理待追加日志的序号 大于 endIndex+1 的情况，可以认为有追加积压，处理要点如下 :
                TimeoutFuture<PushEntryResponse> future = (TimeoutFuture<PushEntryResponse>) pair.getValue();
                //1. 如果挂起时间（排队时间）未超时，则继续检查下一条待追加日志
                if (!future.isTimeOut()) {
                    //直接遍历下一个 请求
                    //不执行 minFastForwardIndex 更新操作
                    continue;
                }

                //2）如果已经超时，说明该日志没有正常写入从节点，则记录其日志序号，
                // 然后向主节点汇报，因为这里是在遍历检查每一条待追加日志，
                // 所以最终需要反馈的是最小超时的日志序号
                if (firstEntryIndex < minFastForwardIndex) {
                    minFastForwardIndex = firstEntryIndex;
                }
            }//for over

            //说明 writeRequestMap 还是没有数据 || firstEntryIndex > endIndex 的数据 未超时
            if (minFastForwardIndex == Long.MAX_VALUE) {
                return;
            }
            //找到 最小的快进 消息 indexId
            //执行到这里说明 存在超时的数据 尝试从 writeRequestMap 再次获取  最小的需要处理的数据
            Pair<PushEntryRequest, CompletableFuture<PushEntryResponse>> pair = writeRequestMap.get(minFastForwardIndex);
            //没有找到说明 被 doWork() 移除并处理了 || ???
            if (pair == null) {
                //直接退出
                return;
            }
            //找到的话就返回 INCONSISTENT_STATE
            logger.warn("[PushFastForward] ledgerEndIndex={} entryIndex={}", endIndex, minFastForwardIndex);
            //第五步：则向主节点报告从节点已经与主节点发生了数据不一致，从节点并没有写入序号 minFastForwardIndex 的日志
            //如果主节点收到此种响应，将会停止日志转发，转而向各个从节点发送 COMPARE 请求，从而使数据恢复一致
            pair.getValue().complete(buildResponse(pair.getKey(), DLedgerResponseCode.INCONSISTENT_STATE.getCode()));
        }

        /**
         * The leader does push entries to follower, and record the pushed index. But in the following conditions, the push may get stopped.
         * * If the follower is abnormally shutdown, its ledger end index may be smaller than before. 从节点故障重启
         * At this time, the leader may push fast-forward entries, and retry all the time.
         * * If the last ack is missed, and no new message is coming in.The leader may retry push the last message, but the follower will ignore it.
         * 从节点可能在故障恢复期间 没有收到主节点的信息
         *
         * @param endIndex
         */
        private void checkAbnormalFuture(long endIndex) {
            //说明刚刚检查完
            if (DLedgerUtils.elapsed(lastCheckFastForwardTimeMs) < 1000) {
                return;
            }
            //更新一下
            lastCheckFastForwardTimeMs = System.currentTimeMillis();
            //说明 当前没有积压的append()请求，因此可以同样明确地判断出主节点没有推送新的日志
            if (writeRequestMap.isEmpty()) {
                return;
            }
            //检测从节点的 writeRequestMap 内的数据indexId  发现 存在重复 indexId 的话 异常及时上报 Leader 异常
            checkAppendFuture(endIndex);
        }

        @Override
        public void doWork() {
            try {
                //用的是 if  并不是 while
                //第一步：如果当前节点的状态不是从节点，则跳出
                if (!memberState.isFollower()) {
                    //不是从从节点
                    //waitForRunning(1);
                    //直接返回
                    return;
                }
                //第二步：如果compareOrTruncateRequests 队列不为空，说明优先处理COMMIT、COMPARE、TRUNCATE等请求
                if (compareOrTruncateRequests.peek() != null) {
                    //这里使用的是peek、poll等非阻塞方法，然后根据请求的类型，调用对应的方法
                    Pair<PushEntryRequest, CompletableFuture<PushEntryResponse>> pair = compareOrTruncateRequests.poll();
                    PreConditions.check(pair != null, DLedgerResponseCode.UNKNOWN);
                    switch (pair.getKey().getType()) {
                        case TRUNCATE:
                            //truncate() 请求响应
                            handleDoTruncate(pair.getKey().getEntry().getIndex(), pair.getKey(), pair.getValue());
                            break;
                        case COMPARE:
                            //compare() 请求响应
                            handleDoCompare(pair.getKey().getEntry().getIndex(), pair.getKey(), pair.getValue());
                            break;
                        case COMMIT:
                            //commit() 请求响应
                            handleDoCommit(pair.getKey().getCommitIndex(), pair.getKey(), pair.getValue());
                            break;
                        default:
                            break;
                    }
                }
                //todo 从节点 处理 append 请求
                else {
                    // 根据当前节点已存储的最大日志序号 计算下一条待写日志的日志序号
                    // 从待写队列中获取日志的处理请求。如果能查找到对应日志的追加请求，则执行 doAppend() 方法追加日志；
                    // 如果从待写队列中没有找到对应的追加请求，则调用checkAbnormalFuture() 检查追加请求是否丢失
                    long nextIndex = dLedgerStore.getLedgerEndIndex() + 1;
                    // 相当于 写-写  操作 [删除操作]
                    Pair<PushEntryRequest, CompletableFuture<PushEntryResponse>> pair = writeRequestMap.remove(nextIndex);
                    //没有找到合适的准确下一条数据  说明什么??? 再检查检查吧 并没有直接 进行阻塞...
                    if (pair == null) {
                        //checkAbnormalFuture() 检查追加请求是否丢失
                        checkAbnormalFuture(dLedgerStore.getLedgerEndIndex());
                        //阻塞
                        waitForRunning(1);
                        //直接退出 函数 执行下一次循环
                        return;
                    }
                    //System.out.println(" writeRequestMap.remove(nextIndex); : " + pair.getKey().toString());// 都为 0  commitIndex=0
                    //找到下一个 顺序 的数据了
                    PushEntryRequest request = pair.getKey();
                    if (request.isBatch()) {
                        handleDoBatchAppend(nextIndex, request, pair.getValue());
                    } else {
                        //从节点 将要 进行添加的准确下一条数据索引
                        //数据实体
                        //响应 future
                        handleDoAppend(nextIndex, request, pair.getValue());
                    }
                }
            } catch (Throwable t) {
                DLedgerEntryPusher.logger.error("Error in {}", getName(), t);
                //有异常了 直接 睡眠100 毫秒
                DLedgerUtils.sleep(100);
            }
        }
    }
}
